package com.example.bootredis.flink;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.logging.Logger;

/**
 * @author: xrp
 * @date: 2022/05/24/17:27
 * @description  订单sink
 */
@Slf4j
public class ListenerOrderSink extends RichSinkFunction<String> {



    private PreparedStatement ps = null;

    private Connection connection = null;
    String driver = "com.mysql.cj.jdbc.Driver";
    String url = "jdbc:mysql://要将监听到的数据同步到哪，另一个数据库地址:端口号/数据库名字?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC";
    String username = "数据库账号";
    String password = "数据库密码";

    @Override
    public void open(Configuration parameters) throws Exception {
//        super.open(parameters);
//        connection = getConn();
//        ps = connection.prepareStatement("insert into 数据库名.表名values (?,?,?,?,?,?,?,?,?,?,?)");
    }
    private Connection getConn() {
        try {
            Class.forName(driver);
            connection = DriverManager.getConnection(url, username, password);
            log.error("数据库连接成功");
        } catch (Exception e) {
            log.error("数据库连接失败");
        }
        return connection;
    }


    @Override
    public void invoke(String p, Context context) throws Exception {
    //TranslateJson  将自定义的返回结果集转为具体实体
        TranslateJson translateJson = JSON.parseObject(p, TranslateJson.class);
        log.info("同步的数据2：{}",translateJson);
        /*if (ConstantValue.CREATE_INFO.equals(translateJson.getOp())) {
            ErpOrder erpOrder = JSON.parseObject(translateJson.getAfter(), ErpOrder.class);
            // 注意： 字段个数需要与表字段个数对应上
            ps.setString(1,erpOrder.getId());
            ps.setString(2,erpOrder.getCode());
            ps.setString(3,erpOrder.getCustomerName());
            ps.setBigDecimal(4,erpOrder.getOrderAmount());
            ps.setString(5,erpOrder.getUserId());
            ps.setString(6,erpOrder.getUserName());
            ps.setString(7,erpOrder.getCreateName());
            ps.setString(8,erpOrder.getCreateDate());
            ps.setInt(9,erpOrder.getStatus());
            ps.setInt(10,erpOrder.getOrderId());
            ps.setString(11,erpOrder.getOrderCode());
            ps.executeUpdate();
        }*/
    }

    @Override
    public void close() throws Exception {
        /*super.close();
        if(connection != null){
            connection.close();
        }
        if (ps != null){
            ps.close();
        }*/
    }


}

